package com.gis.aoitest.window;

import com.gis.aoitest.beans.SensorReading;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

/**
 * @author LnnuUser
 * @create 2021-09-08-下午2:53
 * 统计15秒内各key的最小温度
 */
public class TimeTest01_EventTimeWindow {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // socket文本流
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 9999);

        // 转换成SensorReading类型
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        } )
            // 时间语义
            // 升序数据设置时间戳和watermark
//            .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SensorReading>() {
//                @Override
//                public long extractAscendingTimestamp(SensorReading element) {
//                    return element.getTimestamp() * 1000L;
//                }
//            })
            // 乱序数据设置时间戳和watermark
            .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(1)) {
            @Override
            public long extractTimestamp(SensorReading element) {
                return element.getTimestamp() * 1000L;
            }
        });

        OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late"){};

        // 基于时间的开窗集合，统计15s内，温度的最小值
        SingleOutputStreamOperator<SensorReading> minTempStream = dataStream.keyBy("id")
                //.timeWindow(Time.seconds(15))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .allowedLateness(Time.minutes(1))
//                .sideOutputLateData(outputTag)
                .minBy("temperature");

        minTempStream.print("minTemp");
        minTempStream.getSideOutput(outputTag).print();

        env.execute();
    }
}
